﻿#include "grpc_interface.h"
#include <QMetaObject>
#include <QDebug>

#include <QMetaObject>

GrpcInterface::GrpcInterface(QObject *parent)
    : QObject(parent),
      m_cq(std::make_unique<grpc::CompletionQueue>())
{
}

GrpcInterface::~GrpcInterface()
{
    shutdown();
    // 确保 CompletionQueue 被正确关闭
    if (m_cq) {
        m_cq->Shutdown();
    }
    // 等待工作线程退出
    if (m_cqThread.joinable()) m_cqThread.join();
}

void GrpcInterface::connectToServer(const std::string &serverAddr)
{
    auto channel = grpc::CreateChannel(serverAddr, grpc::InsecureChannelCredentials());
    m_stub = PluginService::NewStub(channel);

    // 你可以选择是否自动启动流
    startStream();
}

void GrpcInterface::sendSyncRequest(const CommonRequest &req)
{
    if (!m_stub) {
        emit streamError(QStringLiteral("gRPC stub 尚未初始化"));
        return;
    }

    grpc::ClientContext ctx;
    ctx.set_deadline(std::chrono::system_clock::now() + std::chrono::seconds(5));

    CommonResponse resp;
    grpc::Status status = m_stub->HandleRequestSync(&ctx, req, &resp);

    if (!status.ok()) {
        QString err = QStringLiteral("同步 RPC 调用失败: %1 (code=%2)")
                .arg(QString::fromStdString(status.error_message()))
                .arg(status.error_code());
        emit streamError(err);
        return;
    }

    std::string serialized;
    if (resp.SerializeToString(&serialized)) {
        emit streamMessage(QByteArray::fromStdString(serialized));
    } else {
        emit streamError(QStringLiteral("同步响应序列化失败"));
    }
}

void GrpcInterface::startStream()
{
    if (!m_stub) {
        emit streamError(QStringLiteral("gRPC stub 尚未初始化"));
        return;
    }

    // 防止重复启动
    if (m_call) {
        emit streamError(QStringLiteral("流已启动"));
        return;
    }

    m_call = std::make_unique<StreamCall>();
    void* startTag = reinterpret_cast<void*>(static_cast<std::uintptr_t>(StreamCall::Tag::START));
    m_cq = std::make_unique<grpc::CompletionQueue>();
    // 创建异步流
    m_call->stream = m_stub->AsyncHandleRequestStream(&m_call->context, m_cq.get(), startTag);

    m_shutdown.store(false);
    if (m_cqThread.joinable()) {
        m_cqThread.join();
    }
    // 启动 CompletionQueue 事件处理线程
    m_cqThread = std::thread(&GrpcInterface::cqThreadMain, this);

    m_connected.store(true);
    emit streamConnected();
}

void GrpcInterface::cqThreadMain()
{
    void* tag;
    bool ok;

    // CompletionQueue::Next() 循环，直到 Shutdown 后返回 false
    while (m_cq->Next(&tag, &ok)) {
        handleRpcEvent(tag, ok);
    }
    // CQ 已被关闭并排空
}

void GrpcInterface::handleRpcEvent(void* tag_ptr, bool ok)
{
    if (!m_call) {
        // 流已被重置，忽略后续事件
        return;
    }

    auto tagVal = reinterpret_cast<std::uintptr_t>(tag_ptr);
    StreamCall::Tag tag = static_cast<StreamCall::Tag>(tagVal);

    switch (tag) {

    case StreamCall::Tag::START:
        if (!ok) {
            // StartCall 失败
            QMetaObject::invokeMethod(this, [this]() {
                emit streamError(QStringLiteral("StartCall 调用失败"));
            }, Qt::QueuedConnection);

            if (!m_shutdown.exchange(true)) {
                if (m_cq) m_cq->Shutdown();
            }
            m_call.reset();
            return;
        }
        // Start 成功 → 请求第一次 Read
        {
            void* readTag = reinterpret_cast<void*>(static_cast<std::uintptr_t>(StreamCall::Tag::READ));
            m_call->stream->Read(&m_call->response, readTag);
        }
        break;

    case StreamCall::Tag::READ:
        if (!ok) {
            // 服务器关闭写端 → 发送 WritesDone（若未发送）
            std::lock_guard<std::mutex> lg(m_writeMutex);
            if (m_call && !m_call->write_done) {
                void* wdTag = reinterpret_cast<void*>(static_cast<std::uintptr_t>(StreamCall::Tag::WRITES_DONE));
                m_call->stream->WritesDone(wdTag);
                m_call->write_done = true;
            }
            // Read 不会再收到数据，等待 FINISH
            return;
        }

        // 收到一条消息
        {
            std::string serialized;
            if (m_call->response.SerializeToString(&serialized)) {
                QByteArray data = QByteArray::fromStdString(serialized);
                QMetaObject::invokeMethod(
                    this, "emitStreamMessage",
                    Qt::QueuedConnection,
                    Q_ARG(QByteArray, data)
                );
            } else {
                QMetaObject::invokeMethod(this, [this]() {
                    emit streamError(QStringLiteral("流响应序列化失败"));
                }, Qt::QueuedConnection);
            }

            // 再次发起 Read
            void* readTag = reinterpret_cast<void*>(static_cast<std::uintptr_t>(StreamCall::Tag::READ));
            m_call->stream->Read(&m_call->response, readTag);
        }
        break;

    case StreamCall::Tag::WRITE:
        {
            std::lock_guard<std::mutex> lg(m_writeMutex);

            if (!ok) {
                // 写失败 → 流即将关闭
                if (m_call && !m_call->write_done) {
                    void* wdTag = reinterpret_cast<void*>(static_cast<std::uintptr_t>(StreamCall::Tag::WRITES_DONE));
                    m_call->stream->WritesDone(wdTag);
                    m_call->write_done = true;
                }
                m_writeInProgress = false;
                return;
            }

            // 正常写成功 → 弹出队列首元素
            if (!m_writeQueue.empty()) {
                m_writeQueue.pop_front();
            }
            m_writeInProgress = false;
        }
        // 尝试继续写下一条
        tryWriteNext();
        break;

    case StreamCall::Tag::WRITES_DONE:
        // 发送 Finish 请求
        {
            void* finishTag = reinterpret_cast<void*>(static_cast<std::uintptr_t>(StreamCall::Tag::FINISH));
            m_call->stream->Finish(&m_call->status, finishTag);
        }
        break;

    case StreamCall::Tag::FINISH:
        {
            grpc::Status status = m_call->status;

            QMetaObject::invokeMethod(this, [this, status]() {
                if (!status.ok()) {
                    emit streamError(QString::fromStdString(status.error_message()));
                }
                emit streamClosed();
            }, Qt::QueuedConnection);

            // 清理流对象
            m_call.reset();

            // 完成后关闭 CompletionQueue
            if (!m_shutdown.exchange(true)) {
                if (m_cq) m_cq->Shutdown();
            }
        }
        break;

    default:
        break;
    }
}

void GrpcInterface::sendStreamMessage(const CommonRequest &req)
{
    std::lock_guard<std::mutex> lg(m_writeMutex);

    if (!m_call || !m_call->stream) {
        emit streamError(QStringLiteral("流尚未启动"));
        return;
    }

    m_writeQueue.push_back(req);
    tryWriteNext_locked();
}

void GrpcInterface::tryWriteNext()
{
    std::lock_guard<std::mutex> lg(m_writeMutex);
    tryWriteNext_locked();
}

void GrpcInterface::tryWriteNext_locked()
{
    if (!m_call || !m_call->stream) return;
    if (m_writeInProgress) return;
    if (m_writeQueue.empty()) return;

    // 标记为正在写
    m_writeInProgress = true;

    const CommonRequest &nextReq = m_writeQueue.front();

    void* writeTag = reinterpret_cast<void*>(static_cast<std::uintptr_t>(StreamCall::Tag::WRITE));
    m_call->stream->Write(nextReq, writeTag);
}

void GrpcInterface::stopStream()
{
    // 保证只执行一次
    if (m_shutdown.exchange(true)) return;

    {
        std::lock_guard<std::mutex> lg(m_writeMutex);
        if (m_call && m_call->stream && !m_call->write_done) {
            void* wdTag = reinterpret_cast<void*>(static_cast<std::uintptr_t>(StreamCall::Tag::WRITES_DONE));
            m_call->stream->WritesDone(wdTag);
            m_call->write_done = true;
        }
    }

    // 关闭 CompletionQueue → cqThreadMain 会自然退出
    if (m_cq) {
        m_cq->Shutdown();
    }

    if (m_cqThread.joinable()) {
        m_cqThread.join();
    }

    m_call.reset();
    m_connected.store(false);

    QMetaObject::invokeMethod(this, [this]() {
        emit streamClosed();
    }, Qt::QueuedConnection);
}

void GrpcInterface::shutdown()
{
    stopStream();
}

void GrpcInterface::emitStreamMessage(const QByteArray& data) {
    emit streamMessage(data);
}
